package com.jugheadzhou.core.zk;

import lombok.Getter;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.Optional;

public class ZkClient {

	public static final String SLASH ="/";
	public static final String ZKLOCK ="zkLock";
	/**
	 * 分布式锁根节点
	 */
	private static final String LOCK_ROOT_PATH = "/lock/";
	@Getter
	private final CuratorFramework client;
	@Getter
	private ZkProperties zkProps;
	
	public ZkClient(ZkProperties zkProps) {
		this.zkProps  =zkProps;
		Builder builder = CuratorFrameworkFactory.builder()
				.namespace(zkProps.getNamespace())
				.connectString(zkProps.getConnectString())
			    .sessionTimeoutMs(zkProps.getSessionTimeoutMs())
			    .connectionTimeoutMs(zkProps.getConnectionTimeoutMs())
				.retryPolicy(zkProps.getRetryPolicy());
		if(zkProps.isNeedACL()) {
			builder.authorization(zkProps.getAuthInfos());
		}
		client = builder.build();
		client.start();
	}

	public String getLockRootPath(String lockKey){
        return new StringBuilder(LOCK_ROOT_PATH).append(this.zkProps.getServiceName()).append(SLASH).append(Optional.ofNullable(lockKey).orElse(ZKLOCK)).toString();
    }

	public InterProcessMutex createInterProcessMutex(String lockKey) {
		return new InterProcessMutex(client, getLockRootPath(lockKey));
	}

	public String createNode(String path, String data) throws Exception {
		path = formatPath(path);
		return client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, (data==null || data.trim().equals(""))?"".getBytes():data.getBytes());
	}
	
	public String createTempNode(String path, String data, boolean sequenctial) throws Exception {
		path = formatPath(path);
		return client.create().creatingParentsIfNeeded().withMode(sequenctial?CreateMode.EPHEMERAL_SEQUENTIAL : CreateMode.EPHEMERAL).forPath(path, (data==null || data.trim().equals(""))?"".getBytes():data.getBytes());
	}
	
	public boolean isExist(String path) throws Exception {
		path = formatPath(path);
		Stat stat = client.checkExists().forPath(path);
		return stat != null;
	}
	
	public void setNodeValue(String path, String data) throws Exception {
		path = formatPath(path);
		if(isExist(path)) {
			client.setData().forPath(path, (data==null || data.trim().equals(""))?"".getBytes():data.getBytes());
		}		
	}
	
	public String getNodeValue(String path) throws Exception {
		path = formatPath(path);
		return new String(client.getData().forPath(path));
	}
	
	public List<String> getNodeChildren(String path) throws Exception {
		path = formatPath(path);
		return client.getChildren().forPath(path);
	}
	
	public void deleteNode(String path) throws Exception {
		if(isExist(path)) {
			client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
		}
	}
	
	public NodeCache watchNode(String watchPath, NodeCacheListener listener) throws Exception {
		watchPath = formatPath(watchPath);
		final NodeCache cache = new NodeCache(client, watchPath);
		cache.start();
		if(listener!=null) {
			cache.getListenable().addListener(listener);
		}
		return cache;
	}
	
	public PathChildrenCache watchNodeChildren(String watchPath, PathChildrenCacheListener listener) throws Exception {
		watchPath = formatPath(watchPath);
		final PathChildrenCache cache = new PathChildrenCache(client, watchPath, true);
		cache.start();
		if(listener!=null) {
			cache.getListenable().addListener(listener);
		}
		return cache;
	}
	
	public void watchNodeTree(String watchPath, TreeCacheListener listener) throws Exception {
		watchPath = formatPath(watchPath);
		TreeCache cache = new TreeCache(client, watchPath);
		cache.start();
		if(listener!=null)
			cache.getListenable().addListener(listener);
	}
	
	public void close() {
		CloseableUtils.closeQuietly(client);
	}
	
	private String formatPath(String path) {
		if(path==null) {
			return SLASH;
		}
		if(!path.startsWith(SLASH)) {
			return SLASH + path;				
		}else {
			return path;
		}
	}
}
